package com.kerrykidz.system.util.sync;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.jms.DeliveryMode;

import org.quartz.JobExecutionContext;
import org.springframework.jdbc.core.RowCallbackHandler;
import org.springframework.scheduling.quartz.QuartzJobBean;

import com.kerrykidz.sync.dao.model.SysDataSyncEntity;

public abstract class AbstractSyncTask extends QuartzJobBean {

	private static final String SELECT_SYS_DATA_SYNC_SQL = "SELECT * FROM SYS_DATA_SYNC WHERE FK_BRAN_ID=? AND EXTRACT_FLAG = ? AND  TABLE_NAME=? ORDER BY SYNC_TIME_S ASC";

	@Override
	protected void executeInternal(JobExecutionContext context) {
		if (SyncParamHelper.EXECUTE_SYNC_TASK) {
			List<SyncBean> syncs = checkCycleTask(context);
			MessageTask message = collectData(syncs);
			if (message != null) {
				String queueNames[] = ActiveMQHelper.getQueueName();
				for (String queueName : queueNames)
					ActiveMQHelper.push2broker(message, queueName, DeliveryMode.PERSISTENT);
			}
		}
	}

	protected abstract List<SyncBean> checkCycleTask(JobExecutionContext context);

	protected void confirm(String id) {
		String sql = "UPDATE SYS_DATA_SYNC SET EXTRACT_FLAG = '1' WHERE PK_DATASYNC_ID=?";
		SqlHelper.executeUpdate(sql.toString(), id);
	}

	private List<SysDataSyncEntity> loadSyncData(List<SyncBean> syncs) {

		final List<SysDataSyncEntity> datas = new ArrayList<SysDataSyncEntity>();

		/**
		 * 排除掉不应该同步的任务
		 */
		for (SyncBean sync : syncs) {
			if (SyncParamHelper.CURRENT_STORE.equals(SyncParamHelper.ZONGBU)) {
				if (sync.getSyncType() == SyncBean.SYNCTYPE_MD)
					continue;
			} else {
				if (sync.getSyncType() == SyncBean.SYNCTYPE_ZB)
					continue;
			}

			SqlHelper.query(SELECT_SYS_DATA_SYNC_SQL, new RowCallbackHandler() {

				@Override
				public void processRow(ResultSet rs) throws SQLException {
					if (rs.getString("PK_DATASYNC_ID") != null) {
						SysDataSyncEntity sync = new SysDataSyncEntity();
						sync.setPkDatasyncId(rs.getString("PK_DATASYNC_ID"));
						sync.setTableName(rs.getString("TABLE_NAME"));
						sync.setSyncTimeS(rs.getTimestamp("SYNC_TIME_S"));
						sync.setSyncTimeE(rs.getTimestamp("SYNC_TIME_E"));
						datas.add(sync);
					}
				}
			}, SyncParamHelper.CURRENT_STORE, SqlHelper.EXTRACT_FLAG_NO, sync.getTableName());
		}

		return datas;
	}

	private Map<String, SyncBean> formatMap(List<SyncBean> syncs) {
		Map<String, SyncBean> map = new HashMap<String, SyncBean>();
		for (SyncBean sync : syncs)
			map.put(sync.getTableName(), sync);
		return map;
	}

	protected MessageTask collectData(List<SyncBean> syncs) {

		List<SysDataSyncEntity> datas = this.loadSyncData(syncs);
		List<TableSyncMessage> messages = new ArrayList<TableSyncMessage>();
		Map<String, SyncBean> map = this.formatMap(syncs);

		for (SysDataSyncEntity bean : datas) {

			SyncBean sync = map.get(bean.getTableName());

			Object params[] = new Object[] { bean.getSyncTimeS(), bean.getSyncTimeE() };

			List<Map<String, Object>> list = SqlHelper.query(sync.getSql(), params);
			List<SyncMessage> syncMessages = new ArrayList<SyncMessage>();

			/**
			 * 是否有关联表
			 */
			boolean hasRelation = sync.getRelations() != null;

			for (Map<String, Object> o : list) {

				List<RelationMessage> relations;

				if (hasRelation) {
					relations = new ArrayList<RelationMessage>();
					for (SyncBean relation : sync.getRelations()) {
						Object primaryValue = o.get(relation.getPrimary());
						List<Map<String, Object>> redatas = SqlHelper.query(relation.getSql(), primaryValue);
						relations.add(new RelationMessage(redatas, relation.getTableName(), relation.getPrimary(),
								(String) primaryValue, relation.getColum()));
					}
				} else
					relations = null;
				syncMessages.add(new DataMessage(sync, (String) o.get(sync.getPrimary()), o, relations));
			}

			if (syncMessages.size() > 0)
				messages.add(new TableSyncMessage(syncMessages));
			confirm(bean.getPkDatasyncId());
		}

		if (messages.size() > 0)
			return new MessageTask(messages);
		else
			return null;
	}

}
